/**
 * Copyright (c) Huawei Technologies Co., Ltd. 2024. All rights reserved.
 *
 * Licensed under the Apache License, Version 2.0 (the "License");
 * you may not use this file except in compliance with the License.
 * You may obtain a copy of the License at
 *
 * http://www.apache.org/licenses/LICENSE-2.0
 *
 * Unless required by applicable law or agreed to in writing, software
 * distributed under the License is distributed on an "AS IS" BASIS,
 * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
 * See the License for the specific language governing permissions and
 * limitations under the License.
 */

/**
 * Description: Defines the worker service processing create buffer process.
 */
#include "datasystem/worker/object_cache/service/worker_oc_service_create_impl.h"

#include "datasystem/common/log/log.h"
#include "datasystem/common/iam/tenant_auth_manager.h"
#include "datasystem/common/perf/perf_manager.h"
#include "datasystem/common/inject/inject_point.h"
#include "datasystem/common/util/format.h"
#include "datasystem/common/util/status_helper.h"
#include "datasystem/common/util/uuid_generator.h"
#include "datasystem/utils/status.h"
#include "datasystem/worker/authenticate.h"

namespace datasystem {
namespace object_cache {

WorkerOcServiceCreateImpl::WorkerOcServiceCreateImpl(WorkerOcServiceCrudParam &initParam, EtcdClusterManager *etcdCM,
                                                     std::shared_ptr<AkSkManager> akSkManager)
    : WorkerOcServiceCrudCommonApi(initParam), etcdCM_(etcdCM), akSkManager_(std::move(akSkManager))
{
}

Status WorkerOcServiceCreateImpl::Create(const CreateReqPb &req, CreateRspPb &resp)
{
    INJECT_POINT("worker.Create.begin");
    workerOperationTimeCost.Clear();
    Timer timer;
    PerfPoint point(PerfKey::WORKER_CREATE_OBJECT);
    AccessRecorder posixPoint(AccessRecorderKey::DS_POSIX_CREATE);
    LOG(INFO) << FormatString(
        "[Create] Receive create meta request, clientId: %s, objectKey: %s, cacheType: %u, size: %zu", req.client_id(),
        req.object_key(), req.cache_type(), req.data_size());
    int64_t remainingTimeMs = reqTimeoutDuration.CalcRealRemainingTime();
    INJECT_POINT_NO_RETURN("WorkerOcServiceCreateImpl.Create.timeoutMs",
                           [&remainingTimeMs]() { remainingTimeMs = -1; });
    if (remainingTimeMs <= 0) {
        return Status(
            StatusCode::K_RPC_DEADLINE_EXCEEDED,
            FormatString("The create request process time has exceeded the request timeout time (remaining: %lld ms)",
                         remainingTimeMs));
    }
    CHECK_FAIL_RETURN_STATUS(etcdCM_ != nullptr, StatusCode::K_NOT_READY, "ETCD cluster manager is not provided.");
    std::string tenantId;
    RETURN_IF_NOT_OK_PRINT_ERROR_MSG(worker::Authenticate(akSkManager_, req, tenantId), "Authenticate failed.");
    Status rc = CreateImpl(tenantId, req.client_id(), req.object_key(), req.data_size(), resp,
                           static_cast<CacheType>(req.cache_type()));
    RequestParam reqParam;
    reqParam.objectKey = req.object_key().substr(0, LOG_OBJECT_KEY_SIZE_LIMIT);
    posixPoint.Record(rc.GetCode(), std::to_string(req.data_size()), reqParam, rc.GetMsg());
    point.Record();
    workerOperationTimeCost.Append("Total Create", timer.ElapsedMilliSecond());
    LOG(INFO) << FormatString("[Client %s] [ObjectKey %s] The operations of worker Create %s", req.client_id(),
                              req.object_key(), workerOperationTimeCost.GetInfo());
    return rc;
}

Status WorkerOcServiceCreateImpl::CreateImpl(const std::string &tenantId, const std::string &clientId,
                                             const std::string &rawObjectKey, size_t dataSize, CreateRspPb &resp,
                                             CacheType cacheType)
{
    auto objectKey = TenantAuthManager::ConstructNamespaceUriWithTenantId(tenantId, rawObjectKey);
    // Check whether the object is sealed.
    std::shared_ptr<SafeObjType> entry;
    CHECK_FAIL_RETURN_STATUS(
        !(objectTable_->Get(objectKey, entry).IsOk() && entry->Get() != nullptr && (*entry)->IsSealed()),
        K_OC_ALREADY_SEALED, "Cannot create sealed object.");

    // Given size, construct shmUnit, generate shm uuid and add client's reference on shmUnit.
    auto shmUnit = std::make_shared<ShmUnit>();
    auto metadataSize = GetMetadataSize();
    RETURN_IF_NOT_OK_PRINT_ERROR_MSG(
        AllocateMemoryForObject(objectKey, dataSize, metadataSize, true, evictionManager_, *shmUnit, cacheType),
        "worker allocate memory failed");

    std::string shmUnitId;
    IndexUuidGenerator(shmIdCounter.fetch_add(1), shmUnitId);
    shmUnit->id = shmUnitId;
    memoryRefTable_->AddShmUnit(clientId, shmUnit);

    // Construct CreateRespPb.
    resp.set_store_fd(shmUnit->GetFd());
    resp.set_mmap_size(shmUnit->GetMmapSize());
    resp.set_offset(shmUnit->GetOffset());
    resp.set_shm_id(shmUnit->GetId());
    resp.set_metadata_size(metadataSize);
    INJECT_POINT("worker.Create.AllocateMemory");
    return Status::OK();
}

Status WorkerOcServiceCreateImpl::MultiCreate(const MultiCreateReqPb &req, MultiCreateRspPb &resp)
{
    CHECK_FAIL_RETURN_STATUS(etcdCM_ != nullptr, StatusCode::K_NOT_READY, "ETCD cluster manager is not provided.");
    std::string tenantId;
    RETURN_IF_NOT_OK_PRINT_ERROR_MSG(worker::Authenticate(akSkManager_, req, tenantId), "Authenticate failed.");
    CHECK_FAIL_RETURN_STATUS_PRINT_ERROR(Validator::IsBatchSizeUnderLimit(req.object_key_size()),
                                         StatusCode::K_INVALID, "invalid object size");
    CHECK_FAIL_RETURN_STATUS(req.object_key_size() == req.data_size_size(), K_INVALID,
                             FormatString("object key count %zu not match with data size count %zu",
                                          req.object_key_size(), req.data_size_size()));
    auto lastRc = Status::OK();
    for (int i = 0; i < req.object_key().size(); i++) {
         const auto &objectKey = req.object_key(i);
        auto dataSize = req.data_size(i);
        // If some buffer create failed, need to rollback, remove shm-unit.
        CreateRspPb subResp;
        Status rc = CreateImpl(tenantId, req.client_id(), objectKey, dataSize, subResp);
        INJECT_POINT("WorkerOCServiceImpl.MultiCreate.Allocate", [&i, &rc](int failedIndex) {
            if (failedIndex == i) {
                rc = Status(StatusCode::K_RUNTIME_ERROR, "Set runtime error");
            }
            return Status::OK();
        });
        resp.mutable_results()->Add(std::move(subResp));
        if (rc.IsError()) {
            lastRc = rc;
            break;
        }
    }
    // Rollback all memory if failed.
    if (lastRc.IsError()) {
        const auto &clientId = req.client_id();
        for (auto &subResp : resp.results()) {
            memoryRefTable_->RemoveShmUnit(clientId, subResp.shm_id());
        }
        resp.Clear();
    }
    return lastRc;
}
}  // namespace object_cache
}  // namespace datasystem
